package com.zzh.partnersys.im.controller;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.zzh.partnersys.common.constant.redis.ChatRedisConstant;
import com.zzh.partnersys.common.enums.ChatTypeEnum;
import com.zzh.partnersys.common.enums.MessageTypeEnum;
import com.zzh.partnersys.common.enums.SenderTypeEnum;
import com.zzh.partnersys.common.enums.SessionTypeEnum;
import com.zzh.partnersys.common.exception.BusinessException;
import com.zzh.partnersys.common.exception.ResultCodeEnum;
import com.zzh.partnersys.common.util.SnowflakeUtil;
import com.zzh.partnersys.im.entity.ChatMessage;
import com.zzh.partnersys.im.entity.ChatMessageContent;
import com.zzh.partnersys.im.entity.ChatSession;
import com.zzh.partnersys.im.entity.request.AiStreamRequest;
import com.zzh.partnersys.im.mapper.ChatMessageContentMapper;
import com.zzh.partnersys.im.mapper.ChatMessageMapper;
import com.zzh.partnersys.im.mapper.ChatSessionMapper;
import com.zzh.partnersys.im.util.RedisSequenceUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Async;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * AI聊天控制器（IM模块）
 * 负责处理AI聊天业务逻辑，通过远程调用AI模块获取大模型响应
 *
 * @author zzh
 */
@Slf4j
@RestController
@RequestMapping("/chat")
public class AiChatController {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RedisSequenceUtil redisSequenceUtil;

    @Value("${ai.service.url:http://localhost:8123}")
    private String aiServiceUrl;

    /**
     * WebClient用于调用AI模块的流式接口
     */
    private WebClient webClient;

    @Resource
    public void setWebClient(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder
                .baseUrl(aiServiceUrl)
                .build();
    }

    @Resource
    private ChatMessageMapper chatMessageMapper;

    @Resource
    private ChatMessageContentMapper chatMessageContentMapper;

    @Resource
    private ChatSessionMapper chatSessionMapper;

    private SseEmitter senderSseEmitter(String data){
        SseEmitter errorEmitter = new SseEmitter(1000L);
        try {
            errorEmitter.send(SseEmitter.event().name("error").data(data));
            errorEmitter.complete();
        } catch (IOException e) {
            log.error("发送错误信息失败", e);
        }
        return errorEmitter;
    }

    /**
     * SSE 流式调用 AI 聊天应用（用户-AI单聊专用）
     * 核心流程：生成唯一序列号 → 消息预处理 → Redis临时存储 → 远程调用AI模块 → 实时推送给前端 → 异步入库
     *
     * @param request 请求对象，包含message、sessionId、assistantId、userId
     * @return SseEmitter 10分钟超时的SSE发射器（流式推送AI回复）
     */
    @PostMapping(value = "/sse_emitter")
    public SseEmitter doChatWithAiSseEmitter(@RequestBody AiStreamRequest request) {
        String message = request.getMessage();
        String sessionId = request.getSessionId();
        Long assistantId = request.getAssistantId();
        Long userId = request.getUserId();

        // 1. 基础参数校验
        if (sessionId == null || assistantId == null || userId == null || message == null || message.trim().isEmpty()) {
            return senderSseEmitter("必填参数不能为空");
        }

        // 1.2 校验会话合法性：查询chat_session表，确认sessionId存在且session_type=1（用户-AI）、assistantId匹配、未删除
//        ChatSession chatSession = chatSessionMapper.selectOne(new LambdaQueryWrapper<ChatSession>()
//                .eq(ChatSession::getId, sessionId)
//                .eq(ChatSession::getSessionType, SessionTypeEnum.USER_TO_AI.getCode()));
//        if (ObjectUtil.isNull(chatSession)){
//            return senderSseEmitter("会话不存在或已删除");
//        }
        // TODO: 1.3 校验消息格式：如果是非文本消息（msg_type≠1），校验content中的fileUrl是否有效

        // 2. 生成消息唯一标识与序列号
        Long messageId = SnowflakeUtil.nextId();
        Long sequence = redisSequenceUtil.generateSequence(ChatTypeEnum.SINGLE_CHAT.getCode(), Long.valueOf(sessionId));
        String requestId = String.format("%d_%s_%d", userId, sessionId, messageId);

        // 3. 消息预处理与Redis临时存储
        // 3.1 解析消息类型：默认文本（msg_type=1），如果message是JSON字符串，解析msg_type字段
        Integer msgType = MessageTypeEnum.TEXT.getCode();
        String contentJson = message;
        try {
            JSONObject msgJson = JSON.parseObject(message);
            if (msgJson.containsKey("msg_type")) {
                msgType = msgJson.getInteger("msg_type");
            }
            if(msgJson.containsKey("text")){
                contentJson = msgJson.getString("text");
            }
        } catch (Exception e) {
            // 解析失败，默认按文本消息处理
            contentJson = JSON.toJSONString(Collections.singletonMap("text", message));
        }

        // 3.2 压缩消息内容：文本消息用GZIP压缩，减少Redis存储开销
//        byte[] compressedContent = msgType.equals(MessageTypeEnum.TEXT.getCode())
//                ? GzipUtil.compress(contentJson)
//                : contentJson.getBytes(StandardCharsets.UTF_8);

        // 存储消息元数据到Redis
        String msgCacheKey = ChatRedisConstant.buildChatMsgKey(sessionId, ChatTypeEnum.SINGLE_CHAT.getCode());
        Map<String, Object> msgMeta = new HashMap<>();
        msgMeta.put("messageId", messageId);
        msgMeta.put("chatType", ChatTypeEnum.SINGLE_CHAT.getCode());
        msgMeta.put("sessionId", sessionId);
        msgMeta.put("assistantId", assistantId);
        msgMeta.put("userId", userId);
        msgMeta.put("senderType", SenderTypeEnum.USER.getCode());
        msgMeta.put("msgType", msgType);
        msgMeta.put("sequence", sequence);
        msgMeta.put("sendTime", System.currentTimeMillis());
        msgMeta.put("isComplete", false);
        String msgMetaJson = JSON.toJSONString(msgMeta);
        redisTemplate.opsForZSet().add(msgCacheKey, msgMetaJson, sequence);

        // 存储消息原始内容
        String contentCacheKey = ChatRedisConstant.buildChatContentKey(messageId);
//        redisTemplate.opsForValue().set(contentCacheKey, compressedContent, 1, TimeUnit.HOURS);
        redisTemplate.opsForValue().set(contentCacheKey, contentJson, 1, TimeUnit.HOURS);

        // 记录流式会话状态
        final String streamStatusKey = ChatRedisConstant.buildChatStreamStatusKey(messageId);
        final String streamContentKey = ChatRedisConstant.buildChatStreamContentKey(messageId);
        redisTemplate.opsForValue().set(streamStatusKey, "RUNNING", 10, TimeUnit.MINUTES);
        redisTemplate.opsForValue().set(streamContentKey, "", 10, TimeUnit.MINUTES);

        // 4. 创建SSE发射器
        SseEmitter sseEmitter = new SseEmitter(600000L);

        // 5. 构建AI模块请求
        AiStreamRequest aiRequest = new AiStreamRequest();
        aiRequest.setMessage(contentJson);
        aiRequest.setAssistantId(assistantId);
        aiRequest.setStream(true);
        aiRequest.setUserId(userId);
        aiRequest.setSessionId(sessionId);

        // 6. 使用WebClient调用AI模块的流式接口
        webClient.post()
                .uri("/ai/chat/stream")
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(aiRequest)
                .retrieve()
                .bodyToFlux(String.class)
                .timeout(Duration.ofMinutes(10))
                .subscribe(
                        // 正常接收流式片段
                        chunk -> {
                            try {
                                redisTemplate.opsForValue().append(streamContentKey, chunk);
                                Map<String, Object> chunkData = new HashMap<>();
                                chunkData.put("messageId", messageId);
                                chunkData.put("sequence", sequence);
                                chunkData.put("chunk", chunk);
                                chunkData.put("isComplete", false);
                                sseEmitter.send(SseEmitter.event()
                                        .name("message")
                                        .id(requestId)
                                        .data(JSON.toJSONString(chunkData)));
                            } catch (IOException e) {
                                redisTemplate.opsForValue().set(streamStatusKey, "FAILED");
                                sseEmitter.completeWithError(e);
                            }
                        },
                        // 异常回调
                        error -> {
                            try {
                                sseEmitter.send(SseEmitter.event()
                                        .name("error")
                                        .data("AI回复异常：" + error.getMessage()));
                            } catch (IOException e) {
                                // 忽略
                            } finally {
                                redisTemplate.opsForValue().set(streamStatusKey, "FAILED");
                                // 直接调用异步保存方法（不完整消息）
                                saveIncompleteMessage(messageId, sessionId, assistantId, userId, sequence);
                                sseEmitter.completeWithError(error);
                            }
                        },
                        // 完成回调
                        () -> {
                            try {
                                Map<String, Object> completeData = new HashMap<>();
                                completeData.put("messageId", messageId);
                                completeData.put("sequence", sequence);
                                completeData.put("isComplete", true);
                                sseEmitter.send(SseEmitter.event()
                                        .name("complete")
                                        .data(JSON.toJSONString(completeData)));
                            } catch (IOException e) {
                                // 忽略
                            } finally {
                                redisTemplate.opsForValue().set(streamStatusKey, "COMPLETE");
                                // 直接调用异步保存方法（完整消息，不阻塞SSE完成）
                                // 获取AI完整回复内容
                                String aiCompleteContent = (String) redisTemplate.opsForValue().get(streamContentKey);
                                saveCompleteMessage(messageId, sessionId, assistantId, userId, sequence, aiCompleteContent);
                                sseEmitter.complete();
                            }
                        }
                );

        // 7. SSE连接关闭回调
        sseEmitter.onCompletion(() -> {
            redisTemplate.delete(streamStatusKey);
        });

        return sseEmitter;
    }

    /**
     * 异步保存完整消息到MySQL
     * 核心：@Transactional 声明式事务，原子入库用户消息+AI消息+更新会话
     */
    @Async
    @Transactional(rollbackFor = Exception.class)
    public void saveCompleteMessage(Long messageId, String sessionId, Long assistantId, Long userId, Long userSequence) {
        saveCompleteMessage(messageId, sessionId, assistantId, userId, userSequence, null);
    }

    /**
     * 异步保存完整消息到MySQL（带AI内容参数）
     */
    @Async
    @Transactional(rollbackFor = Exception.class)
    public void saveCompleteMessage(Long messageId, String sessionId, Long assistantId, Long userId, Long userSequence, String aiCompleteContent) {
        try {
            // 1. 从Redis获取完整数据
            if (StrUtil.isEmpty(aiCompleteContent)) {
                String streamContentKeyForGet = ChatRedisConstant.buildChatStreamContentKey(messageId);
                aiCompleteContent = (String) redisTemplate.opsForValue().get(streamContentKeyForGet);
                if (StrUtil.isEmpty(aiCompleteContent)) {
                    throw new RuntimeException("AI完整回复为空，入库失败");
                }
            }

            String contentCacheKey = ChatRedisConstant.buildChatContentKey(messageId);
            String userCompressedContent = (String) redisTemplate.opsForValue().get(contentCacheKey);
            if (StrUtil.isEmpty(userCompressedContent)) {
                throw new RuntimeException("用户消息内容为空，入库失败");
            }

            // 从Redis获取用户消息元数据
            String msgMetaKey = ChatRedisConstant.buildChatMsgKey(sessionId, ChatTypeEnum.SINGLE_CHAT.getCode());
            Set<Object> userMsgMetaSet = redisTemplate.opsForZSet().rangeByScore(msgMetaKey, userSequence.doubleValue(), userSequence.doubleValue());
            if (CollUtil.isEmpty(userMsgMetaSet)) {
                throw new RuntimeException("Redis中未找到用户消息元数据，messageId=" + messageId);
            }
            String userMsgMetaJson = userMsgMetaSet.iterator().next().toString();
            JSONObject userMsgMeta = JSON.parseObject(userMsgMetaJson);
            Integer userMsgType = userMsgMeta.getInteger("msgType");
            Long sendTime = userMsgMeta.getLong("sendTime");

            // 2. 入库chat_message_content
            ChatMessageContent userContentPO = new ChatMessageContent();
            userContentPO.setContent(userCompressedContent); // Base64编码后存储
            chatMessageContentMapper.insert(userContentPO);
            Long userContentId = userContentPO.getId();

//            byte[] aiCompressedContent = GzipUtil.compress(aiCompleteContent);
            ChatMessageContent aiContentPO = new ChatMessageContent();
            aiContentPO.setContent(userCompressedContent);
            chatMessageContentMapper.insert(aiContentPO);
            Long aiContentId = aiContentPO.getId();

            // 3. 生成AI消息的序列号
            Long aiSequence = redisSequenceUtil.generateSequence(ChatTypeEnum.SINGLE_CHAT.getCode(), Long.valueOf(sessionId));

            // 4. 入库用户消息元数据
            ChatMessage userMessagePO = new ChatMessage();
            userMessagePO.setId(messageId);
            userMessagePO.setChatType(ChatTypeEnum.SINGLE_CHAT.getCode());
            userMessagePO.setTargetId(Long.valueOf(sessionId));
            userMessagePO.setAssistantId(assistantId);
            userMessagePO.setSenderId(userId);
            userMessagePO.setSenderType(SenderTypeEnum.USER.getCode());
            userMessagePO.setReceiverId(0L);
            userMessagePO.setMsgType(userMsgType);
            userMessagePO.setContentId(userContentId);
            userMessagePO.setSequence(userSequence.intValue());
            userMessagePO.setIsComplete(1);
            userMessagePO.setSendTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(sendTime), ZoneId.systemDefault()));
            userMessagePO.setIsDelete(0);
            chatMessageMapper.insert(userMessagePO);

            // 5. 入库AI消息元数据
            com.zzh.partnersys.im.entity.ChatMessage aiMessagePO = new com.zzh.partnersys.im.entity.ChatMessage();
            aiMessagePO.setId(SnowflakeUtil.nextId());
            aiMessagePO.setChatType(ChatTypeEnum.SINGLE_CHAT.getCode());
            aiMessagePO.setTargetId(Long.valueOf(sessionId));
            aiMessagePO.setAssistantId(assistantId);
            aiMessagePO.setSenderId(0L);
            aiMessagePO.setSenderType(SenderTypeEnum.AI.getCode());
            aiMessagePO.setReceiverId(userId);
            aiMessagePO.setMsgType(1);
            aiMessagePO.setContentId(aiContentId);
            aiMessagePO.setSequence(aiSequence.intValue());
            aiMessagePO.setIsComplete(1);
            aiMessagePO.setSendTime(LocalDateTime.now());
            aiMessagePO.setIsDelete(0);
            chatMessageMapper.insert(aiMessagePO);

            // 6. 更新会话最后消息时间
            ChatSession session = new ChatSession();
            session.setId(Long.valueOf(sessionId));
            session.setLastMsgTime(LocalDateTime.now());
            session.setLastReadSequence(aiSequence.intValue());
            chatSessionMapper.updateById(session);

            // 7. 清理Redis临时数据
            String streamContentKeyForDelete = ChatRedisConstant.buildChatStreamContentKey(messageId);
                redisTemplate.delete(streamContentKeyForDelete);
            redisTemplate.delete(contentCacheKey);

            log.info("完整消息入库成功，messageId: {}, sessionId: {}, userId: {}", messageId, sessionId, userId);
        } catch (Exception e) {
            log.error("完整消息入库失败，messageId={}", messageId, e);
            redisTemplate.opsForValue().set(
                    ChatRedisConstant.buildChatDbTaskKey(messageId),
                    "FAILED",
                    24,
                    TimeUnit.HOURS
            );
            throw new RuntimeException("完整消息入库失败", e);
        }
    }

    /**
     * 异步保存不完整消息到MySQL
     * 核心：仅入库用户消息，AI消息标记为不完整，支持后续重试
     */
    @Async
    @Transactional(rollbackFor = Exception.class)
    public void saveIncompleteMessage(Long messageId, String sessionId, Long assistantId, Long userId, Long userSequence) {
        try {
            // 1. 从Redis获取用户消息内容
            String contentCacheKey = ChatRedisConstant.buildChatContentKey(messageId);
            String userCompressedContent = (String) redisTemplate.opsForValue().get(contentCacheKey);
            if (StrUtil.isEmpty(userCompressedContent)) {
                throw new BusinessException("用户消息内容为空，入库失败", ResultCodeEnum.PARAMS_ERROR.getCode());
            }

            // 从Redis获取用户消息元数据
            String msgMetaKey = ChatRedisConstant.buildChatMsgKey(sessionId, ChatTypeEnum.SINGLE_CHAT.getCode());
            Set<Object> userMsgMetaSet = redisTemplate.opsForZSet().rangeByScore(msgMetaKey, userSequence.doubleValue(), userSequence.doubleValue());
            if (CollUtil.isEmpty(userMsgMetaSet)) {
                throw new BusinessException("Redis中未找到用户消息元数据，messageId=" + messageId, ResultCodeEnum.PARAMS_ERROR.getCode());
            }
            String userMsgMetaJson = userMsgMetaSet.iterator().next().toString();
            JSONObject userMsgMeta = JSON.parseObject(userMsgMetaJson);
            Integer userMsgType = userMsgMeta.getInteger("msgType");
            Long sendTime = userMsgMeta.getLong("sendTime");

            // 2. 入库用户消息内容
            ChatMessageContent userContentPO = new ChatMessageContent();
            userContentPO.setContent(userCompressedContent);
            chatMessageContentMapper.insert(userContentPO);
            Long userContentId = userContentPO.getId();

            // 3. 入库用户消息元数据（AI消息不入库，标记为不完整）
            ChatMessage userMessagePO = new ChatMessage();
            userMessagePO.setId(messageId);
            userMessagePO.setChatType(ChatTypeEnum.SINGLE_CHAT.getCode());
            userMessagePO.setTargetId(Long.valueOf(sessionId));
            userMessagePO.setAssistantId(assistantId);
            userMessagePO.setSenderId(userId);
            userMessagePO.setSenderType(SenderTypeEnum.USER.getCode());
            userMessagePO.setReceiverId(0L);
            userMessagePO.setMsgType(userMsgType);
            userMessagePO.setContentId(userContentId);
            userMessagePO.setSequence(userSequence.intValue());
            userMessagePO.setIsComplete(1); // 用户消息完整
            userMessagePO.setSendTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(sendTime), ZoneId.systemDefault()));
            userMessagePO.setIsDelete(0);
            chatMessageMapper.insert(userMessagePO);

            // 清理Redis临时数据
            redisTemplate.delete(contentCacheKey);

            log.info("不完整消息入库成功，messageId: {}, sessionId: {}, userId: {}", messageId, sessionId, userId);
        } catch (Exception e) {
            log.error("不完整消息入库失败，messageId={}", messageId, e);
            redisTemplate.opsForValue().set(
                    ChatRedisConstant.buildChatDbTaskKey(messageId),
                    "FAILED",
                    24,
                    TimeUnit.HOURS
            );
            throw new BusinessException("不完整消息入库失败", ResultCodeEnum.PARAMS_ERROR.getCode());
        }
    }
}

